//
// Created by 86134 on 2024/1/18.
//

#include "TcpBuffer.hpp"
#include "../base/Defind.hpp"
#include "../base/BaseLog.hpp"
#include "BlockPool.hpp"

#include <cstddef>
#include <functional>
#include <cstring>
#include <unistd.h>
#include <utility>

using namespace DBServer::Socket;


static int Read(int fd, char *buffer) {
    if (BUFFER_SIZE > SIZE_MAX)
    {
        error("what(): configure BUFFER_SIZE is too big!\n");
        exit(1);
    } else if (BUFFER_SIZE <= 0) {
        error("what(): configure BUFFER_SIZE <= 0!\n");
        exit(1);     
    }
    // read the data fom the kernel tcp buffer
    int ret = read(fd, buffer, BUFFER_SIZE);
    if (ret == -1)
    {
        if (errno == EINTR)
        {
            // if it interrupted by a singal
            // read again
            ret = read(fd, buffer + strlen(buffer), BUFFER_SIZE) + ret;
        }
        else if (errno == EAGAIN || errno == EWOULDBLOCK)
        {
            return 0;
        }
        else
        {
            error("what(): read() fail!\n");
            return -1;
        }        
    }
    return ret;
}

static int Write(int fd, std::pair<MemoryBlock*, size_t>& data) {
    int ret;
    try
    {
        ret = write(fd, data.first, data.second);
    }
    catch(const std::exception& e)
    {
        error(e.what());
        return -1;
    }
    return ret;
}

#if 0
#include <unistd.h>
#include <fcntl.h>

int main() {
    int fd[2] = {0};
    pipe(fd);
    char buff[1024];
    char r[1024];
    // write 
    for (int i = 0; i < 10; i++) {
        short len = (i+1) * 10;
        for (int j = 0; j < len; j++)
            buff[j+2] = 'a'+i;
        memcpy(buff, &len, sizeof(len));
        write(fd[1], buff, len+2);
        memset(buff, 0, 1024);
    }
    Node *node = new Node();
    auto v = node->analize(fd[0]);
    for (auto s : v) std::cout << s << std::endl;
    std::cout << "finish" << std::endl;
}

#endif


/*----------------------------------------------------------
 * 
 *           Buffer
 * 
 * ----------------------------------------------------------
*/

TcpBuffer::TcpBuffer() 
    : data_len(0),
      data(nullptr),
      cur_len_head(0), 
      cur_len_data(0), 
      status(HEAD_LACK)
{
    buffer = MemoryPool::GetInstance()->New();
}

TcpBuffer::~TcpBuffer() {
    if (data != nullptr) delete[] data;
    MemoryPool::GetInstance()->Delete(buffer);
}

std::vector<std::shared_ptr<Request>> TcpBuffer::read(int fd, int& res) {
    std::vector<std::pair<char*, short>> request_set = analize(fd, res);
    std::vector<std::shared_ptr<Request>> req_set;
    for (int i = 0; i < (int)request_set.size(); i++)
    {
        // create a request
        std::shared_ptr<Request> req = std::make_shared<Request>(request_set[i].first, request_set[i].second);
        req_set.push_back(std::move(req));

    }
    return req_set;
}


void TcpBuffer::write(std::string& respond) {
    // find  and construct a sending message [len + uuid + data]
    short len = respond.size();
    // limit the data len
    if (len > SHORT_MAX) 
    {
        error("what(): the data len is overflow\n");
        return;
    }

    std::queue<std::pair<MemoryBlock*, size_t>> finish_queue;
    MemoryBlock *write_buff = MemoryPool::GetInstance()->New();
    int buff_offset = 0;

    // when buff is not enough to copy call this function
    std::function<void(const void*, int)> copy_and_new = [&](const void *data, int need) {
        memcpy(write_buff + buff_offset, data, need);
        // push into the consume_queue(local) ensuer message order
        finish_queue.push({write_buff, MEMORY_POOL_BLOCK_SIZE});
        // get a new empty buff
        write_buff = MemoryPool::GetInstance()->New();
        buff_offset = 0;
    };

    // copy the head
    int need = MEMORY_POOL_BLOCK_SIZE - buff_offset;
    if (need < HEAD_LEN)
        copy_and_new(&len, need);
    need = (need > HEAD_LEN) ? 0 : need;
    memcpy(write_buff + buff_offset, &len + need, HEAD_LEN - need);
    buff_offset += (HEAD_LEN - need);

    // copy the data
    const char *data = respond.c_str();
    int offset = 0;     // data offset
    while ((need = MEMORY_POOL_BLOCK_SIZE - buff_offset) < len)
    {
        copy_and_new(data + offset, need);
        len -= need;
        offset += need;
    }
    need = (need > len) ? 0 : need;
    memcpy(write_buff + buff_offset, data + offset + need, len - need);
    buff_offset += len;
    // push the last data
    finish_queue.push({write_buff, buff_offset});
    write_buff = MemoryPool::GetInstance()->New();

    { 
        // push into consume_queue
        std::unique_lock<std::mutex> lock(this->write_mtx);
        while (!finish_queue.empty())
        {
            consume_queue.push(std::move(finish_queue.front()));
            finish_queue.pop();
        }
    }
}

int TcpBuffer::push(int fd) {
    // the block to write
    int push_num = consume_queue.size();
    // write to tco kernel buffer
    for (int i = 0 ; i < push_num; i++)
    {
        auto &data = consume_queue.front();
        if (Write(fd, data) == -1)
        {
            // socket error
            error("what() read error, socket error\n");
            return -1;
        }
        // back memory block to memory pool
        MemoryPool::GetInstance()->Delete(data.first);
        consume_queue.pop();
    }
}

std::vector<std::pair<char*, short>>
TcpBuffer::analize(int fd, int& ret) {
    std::vector<std::pair<char*, short>> res;
    // read the data from kernel tcp buffer
    int total_len = Read(fd, this->buffer);
    if (total_len == -1) 
    {
        // socket is error
        ret = -1;
        return {};
    }
    int read_len = 0;

    while (total_len > 0)
    {
        switch (status)
        {
        case HEAD_LACK:
        {
            if (HEAD_LEN - cur_len_head > total_len)    // head data lack
            {
                memcpy(&data_len + cur_len_head, buffer + read_len, total_len);
                cur_len_head += total_len;
                total_len = 0;
                status = HEAD_LACK;
            } 
            else
            {
                int need = HEAD_LEN - cur_len_head;
                memcpy(&data_len + cur_len_head, buffer + read_len, need);
                cur_len_head += need;
                total_len -= need;
                read_len += need;
                data = new char[data_len+1];
                status = DATA_LACK;
            }
            break;
        }
        case DATA_LACK:
        {
            if (total_len < data_len)
            {
                memcpy(data + cur_len_data, buffer + read_len, total_len);
                cur_len_data += total_len;
                total_len = 0;
                status = DATA_LACK;
            } 
            else
            {
                int need  = data_len - cur_len_data;
                memcpy(data + cur_len_data, buffer + read_len, need);
                total_len -= need;
                read_len += need;
                cur_len_data += need;
                // push into the result set
                data[data_len] = '\0'; // c string to build a std::string
                res.push_back({data, data_len+1});
                // reset the attribution of node
                data = nullptr;
                data_len = 0;
                cur_len_data = 0;
                cur_len_head = 0;
                status = HEAD_LACK;
            }
            break;
        }
        default:
            break;
        }
    }
    memset(buffer, 0, MEMORY_POOL_BLOCK_SIZE);
    return res;
}